package com.xf.kafkatest;

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

public class MyProducer {
    public static void main(String[] args) {

        // Step 1: 配置生产者属性
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "hadoop01:9092,hadoop02:9092,hadoop03:9092"); // Kafka 服务器地址
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.ACKS_CONFIG, "all"); // 确保所有副本都确认
        props.put(ProducerConfig.RETRIES_CONFIG, 3); // 重试次数

        // Step 2: 创建 KafkaProducer 实例
        KafkaProducer<String, String> producer = new KafkaProducer<>(props);

        // Step 3: 发送消息
        for (int i = 0; i < 10; i++) { // 示例发送10条消息
            String key = "key-" + i;
            String value = "Hello Kafka! Message " + i;

            ProducerRecord<String, String> record = new ProducerRecord<>("disTopic", key, value);

            // 异步发送消息（推荐方式）
            producer.send(record, (metadata, exception) -> {
                if (exception == null) {
                    System.out.println("Message sent successfully: topic=" + metadata.topic() +
                            ", partition=" + metadata.partition() +
                            ", offset=" + metadata.offset());
                } else {
                    System.err.println("Error sending message: " + exception.getMessage());
                }
            });
        }

        // 关闭生产者
        producer.close();

        System.out.println("All messages sent and producer closed.");

    }
}
